//go:build linux
// +build linux

package store

import (
	"encoding/binary"
	"fmt"
	"github.com/hashicorp/raft"
	"log"
	"os"
	"reflect"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"syscall"
	"time"
	"unsafe"
)

const dbSuffix = ".mdb"

// 1GB
const defaultSegmentSize = 1 * 1024 * 1024 * 1024

// MinSyncInterval 100 us
const MinSyncInterval = 100 * time.Microsecond

var EmptyByte = make([]byte, 0)

var isLittleEndian = IsLittleEndian()
var pageSize = uint64(os.Getpagesize())

// MMapLog todo: 周期性调度sync刷盘
type MMapLog struct {
	sync.RWMutex
	syncLock *sync.RWMutex

	// index to meta
	ids     []int
	offsets []int
	lens    []int

	// m-map data
	data [][]byte

	dir         string
	segmentSize uint64

	id      int
	address uint64
	synced  *uint64
	cached  *uint64

	// increment unique index
	last       *uint64
	syncRunner *SyncRunner
}

func NewLog(dir string) *MMapLog {
	return NewLogWithSegmentSize(dir, defaultSegmentSize)
}

func NewLogWithSegmentSize(dir string, segmentSize int64) *MMapLog {
	mapLog := &MMapLog{
		syncLock: new(sync.RWMutex),

		ids:     make([]int, 0, 1024),
		offsets: make([]int, 0, 1024),
		lens:    make([]int, 0, 1024),
		data:    make([][]byte, 1, 1024),

		dir:         dir,
		segmentSize: uint64(segmentSize),

		id:     0,
		synced: new(uint64),
		cached: new(uint64),

		last: new(uint64),
	}
	syncRunner := NewSyncRunner(mapLog, MinSyncInterval)
	mapLog.syncRunner = syncRunner
	err := InitFiles(mapLog)
	if err != nil {
		panic(err)
	}
	log.Println(fmt.Sprintf("loading log id=%d, last=%d", mapLog.id, atomic.LoadUint64(mapLog.last)))
	return mapLog
}

func InitFiles(mf *MMapLog) error {
	err := MakeDir(mf.dir)
	if err != nil {
		return nil
	}
	dbs, err := listDbFiles(mf.dir)
	if err != nil {
		return err
	}
	size := len(dbs)
	if size >= 1 {
		sort.Strings(dbs)
		for i, db := range dbs {
			name := fmt.Sprintf("%s/%s", mf.dir, db)
			file, err2 := os.OpenFile(name, os.O_CREATE|os.O_RDWR, os.ModePerm)
			if err2 != nil {
				return err
			}
			data := OpenMapper(file, int(mf.segmentSize))
			mf.data = append(mf.data, data)
			last := recoverData(mf, data, i+1)
			mf.id = i + 1
			mf.address = Address(data)
			atomic.StoreUint64(mf.synced, last)
			atomic.StoreUint64(mf.cached, last)
		}
	} else {
		err1 := mf.NewFile()
		if err1 != nil {
			return err1
		}
	}
	return nil
}

func recoverData(mf *MMapLog, data []byte, id int) uint64 {
	pos := 0
	for {
		target := pos
		term := binary.LittleEndian.Uint32(data[target:])
		if term == 0 {
			return uint64(pos)
		}
		target += 8
		length := binary.LittleEndian.Uint32(data[target:])
		target += 4
		index := binary.LittleEndian.Uint64(data[target:])
		target += 8
		target += 4
		target += int(length)
		mf.saveIndex(int(index), id, pos, target-pos)
		mf.saveLast(index)
		pos = target
	}
}

func (m *MMapLog) NewFile() error {
	// sync before file
	if m.id != 0 {
		err := m.Sync()
		if err != nil {
			panic(err)
		}
	}
	m.id++
	name := fmt.Sprintf("%s/%09d%s", m.dir, m.id, dbSuffix)
	file, err := os.OpenFile(name, os.O_CREATE|os.O_RDWR, os.ModePerm)
	if err != nil {
		return err
	}
	// fast extend file size to segmentSize
	fill := make([]byte, pageSize)
	_, err = file.WriteAt(fill, int64(m.segmentSize-pageSize))
	if err != nil {
		return err
	}
	err = file.Sync()
	if err != nil {
		return err
	}
	data := OpenMapper(file, int(m.segmentSize))
	m.data = append(m.data, data)
	m.address = Address(data)
	atomic.StoreUint64(m.synced, 0)
	atomic.StoreUint64(m.cached, 0)
	return nil
}

func OpenMapper(file *os.File, length int) []byte {
	if length <= 0 {
		return EmptyByte
	}
	b, err := syscall.Mmap(int(file.Fd()), 0, length, syscall.PROT_WRITE|syscall.PROT_READ, syscall.MAP_SHARED)
	if err != nil {
		panic(err)
	}
	return b
}

func (m *MMapLog) FirstIndex() (uint64, error) {
	return 1, nil
}

func (m *MMapLog) LastIndex() (uint64, error) {
	return atomic.LoadUint64(m.last), nil
}

func (m *MMapLog) Last() uint64 {
	return atomic.LoadUint64(m.last)
}

func (m *MMapLog) GetLog(index uint64, raftLog *raft.Log) error {
	if index > atomic.LoadUint64(m.last) {
		return raft.ErrLogNotFound
	}
	id := m.ids[index]
	offset := m.offsets[index]
	size := m.lens[index]
	if size == 0 {
		return raft.ErrLogNotFound
	}
	data := make([]byte, size)
	m.RLock()
	copy(data, m.data[id][offset:offset+size])
	m.RUnlock()
	pos := 0
	raftLog.Term = binary.LittleEndian.Uint64(data[pos:])
	pos += 8
	length := binary.LittleEndian.Uint32(data[pos:])
	pos += 4
	raftLog.Index = binary.LittleEndian.Uint64(data[pos:])
	pos += 8
	raftLog.Type = raft.LogType(binary.LittleEndian.Uint32(data[pos:]))
	pos += 4
	if length != 0 {
		raftLog.Data = data[pos : pos+int(length)]
	} else {
		raftLog.Data = nil
	}
	return nil
}

func (m *MMapLog) StoreLog(raftLog *raft.Log) error {
	_ = m.saveLog(raftLog)
	m.syncRunner.WeakUp()
	return nil
}

func (m *MMapLog) saveLog(raftLog *raft.Log) error {
	dataLen := len(raftLog.Data)
	size := uint64(dataLen + 8 + 8 + 8)

	// 先把数据准备好
	data := make([]byte, size)
	pos := 0
	binary.LittleEndian.PutUint64(data[pos:], raftLog.Term)
	pos += 8
	binary.LittleEndian.PutUint32(data[pos:], uint32(dataLen))
	pos += 4
	binary.LittleEndian.PutUint64(data[pos:], raftLog.Index)
	pos += 8
	binary.LittleEndian.PutUint32(data[pos:], uint32(raftLog.Type))
	pos += 4
	if dataLen != 0 {
		copy(data[pos:pos+dataLen], raftLog.Data[0:])
	}

	// sync的时候,会获取 cached, 要保证cached的时候data已经复制给mmap了
	m.syncLock.Lock()
	end := atomic.AddUint64(m.cached, size)
	id := m.id
	if end >= m.segmentSize-4 {
		m.Lock()
		end = atomic.LoadUint64(m.cached)
		id = m.id
		// double check
		if end >= m.segmentSize-4 {
			err := m.NewFile()
			if err != nil {
				panic(err)
			}
		}
		end = atomic.AddUint64(m.cached, size)
		id = m.id
		m.Unlock()
	}
	start := end - size
	copy(m.data[id][start:end], data)
	m.syncLock.Unlock()
	// sync的时候,会获取 cached, 要保证written的时候data已经复制给mmap了
	m.saveIndex(int(raftLog.Index), id, int(start), int(size))
	m.saveLast(raftLog.Index)
	return nil
}

func (m *MMapLog) saveLast(index uint64) {
	if index > atomic.LoadUint64(m.last) {
		m.Lock()
		if index > atomic.LoadUint64(m.last) {
			atomic.StoreUint64(m.last, index)
		}
		m.Unlock()
	}
}

func (m *MMapLog) saveIndex(index int, id int, pos int, size int) {
	m.Lock()
	if index >= len(m.ids) {
		// todo: 快速扩容数组
		for index >= len(m.ids) {
			m.ids = append(m.ids, 0)
			m.offsets = append(m.offsets, 0)
			m.lens = append(m.lens, 0)
		}
	}
	m.ids[index] = id
	m.offsets[index] = pos
	m.lens[index] = size
	m.Unlock()
}

func (m *MMapLog) StoreLogs(logs []*raft.Log) error {
	for i := range logs {
		_ = m.saveLog(logs[i])
	}
	m.syncRunner.WeakUp()
	return nil
}

func (m *MMapLog) Sync() error {
	// 原子操作: 获取需要flush的page
	m.syncLock.RLock()
	cached := atomic.LoadUint64(m.cached)
	synced := atomic.LoadUint64(m.synced)
	address := m.address
	if synced == cached {
		m.syncLock.RUnlock()
		return nil
	}
	atomic.StoreUint64(m.synced, cached)
	m.syncLock.RUnlock()

	// 执行刷盘
	start := synced / pageSize * pageSize
	length := cached - start

	var ptr uint64
	if isLittleEndian {
		ptr = address + start
	} else {
		ptr = address - start
	}
	_, _, err := syscall.Syscall(syscall.SYS_MSYNC, uintptr(ptr), uintptr(length), syscall.MS_SYNC)
	if err != 0 {
		panic(err)
	}
	return nil
}

func (m *MMapLog) DeleteRange(min, max uint64) error {
	return nil
}

func (m *MMapLog) WaitFlush(index uint64) {
	m.syncRunner.WaitDone(index)
}

func (m *MMapLog) close() {
	for i := range m.data {
		if m.data[i] != nil {
			syscall.Munmap(m.data[i])
		}
	}
}

func MakeDir(dir string) error {
	_, err := os.Stat(dir)
	if err != nil && !os.IsNotExist(err) {
		return err
	}

	if os.IsNotExist(err) {
		err = os.MkdirAll(dir, os.ModePerm)
		if err != nil {
			return err
		}
	}
	return nil
}

func listDbFiles(dir string) ([]string, error) {
	dirFp, err := os.OpenFile(dir, os.O_RDONLY, os.ModeDir)
	if err != nil {
		return nil, err
	}
	defer dirFp.Close()
	lists, err := dirFp.Readdirnames(-1)
	if err != nil {
		return nil, err
	}
	var logLists []string
	for _, v := range lists {
		if strings.HasSuffix(v, dbSuffix) {
			logLists = append(logLists, v)
		}
	}
	return logLists, nil
}

func Address(data []byte) uint64 {
	hdr := (*reflect.SliceHeader)(unsafe.Pointer(&data))
	return uint64(hdr.Data)
}

func IsLittleEndian() bool {
	var value int32 = 1 // 占4byte 转换成16进制 0x00 00 00 01
	// 大端(16进制)：00 00 00 01
	// 小端(16进制)：01 00 00 00
	pointer := unsafe.Pointer(&value)
	pb := (*byte)(pointer)
	if *pb != 1 {
		return false
	}
	return true
}
